package com.yunxiao.cloud.oss.hdfs;

import com.yunxiao.cloud.oss.constant.OssConstant;
import com.yunxiao.cloud.oss.property.HadoopProperties;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @description: HadoopClient
 * @date: 2019/12/17 14:47
 * @author: yunxiao
 * @version: 1.0
 */
public class HadoopClient {
    private static final Logger logger = LoggerFactory.getLogger(HadoopClient.class);

    private FileSystem fileSystem;
    private HadoopProperties hadoopProperties;

    public HadoopClient(FileSystem fileSystem, HadoopProperties hadoopProperties) {
        this.fileSystem = fileSystem;
        this.hadoopProperties = hadoopProperties;
    }

    /**
     * 创建目录
     *
     * @param folderPath 文件夹路径名称
     * @param create     不存在的话是否新建目录
     * @return 是否创建成功
     */
    public boolean mkdir(String folderPath, boolean create) {
        logger.info("【开始创建目录】 文件夹路径名称: {}", folderPath);
        boolean flag = false;
        if (StringUtils.isEmpty(folderPath)) {
            return flag;
        }

        try {
            Path path = new Path(folderPath);
            if (create && !fileSystem.exists(path)) {
                fileSystem.mkdirs(path);
            }
            if (fileSystem.getFileStatus(path).isDirectory()) {
                flag = true;
            }
        } catch (Exception e) {
            logger.error("【创建目录失败】", e);
        }
        return flag;
    }


    /**
     * 文件上传
     *
     * @param delSrc    指是否删除源文件，true为删除，默认为false
     * @param overwrite 是否覆盖
     * @param srcFile   源文件
     * @param destPath  fs的目标路径
     */
    public boolean copyFileToHdfs(boolean delSrc, boolean overwrite, String srcFile, String destPath) {
        boolean result = false;
        logger.info("【文件上传】 开始上传, 上传文件路径: {}, 上传文件:{}", destPath, srcFile);
        Path srcPath = new Path(srcFile);
        // 目标路径
        if (hadoopProperties.getNameNode() != null) {
            destPath = hadoopProperties.getNameNode() + destPath;
        }

        Path dstPath = new Path(destPath);
        try {
            if (!fileSystem.exists(dstPath)) {
                fileSystem.mkdirs(dstPath);
            }

            fileSystem.copyFromLocalFile(delSrc, overwrite, srcPath, dstPath);
            result = true;
        } catch (IOException e) {
            logger.error("【文件上传失败】", e);
        }
        return result;
    }

    /**
     * 删除文件或者文件目录
     *
     * @param path     文件目录路径
     * @param fileName 文件名称
     */
    public boolean rmdir(String path, String fileName) {
        logger.info("【删除文件】 开始删除, 删除文件目录的路径: {}, 文件名称: {}", path, fileName);
        boolean result = false;
        try {
            if (StringUtils.isNotBlank(fileName)) {
                path = path + OssConstant.FILE_SEPARATION + fileName;
            }
            // 删除文件或者文件目录
            result = fileSystem.delete(new Path(path), true);
        } catch (Exception e) {
            logger.error("【删除文件失败】", e);
        }
        return result;
    }

    /**
     * 下载文件
     *
     * @param path         路径
     * @param fileName     文件名称
     * @param outputStream 输出流
     */
    public boolean download(String path, String fileName, OutputStream outputStream) {
        logger.info("【下载文件】 开始下载, 下载文件名称: {}", fileName);
        boolean result = false;
        InputStream inputStream = null;
        try {
            inputStream = fileSystem.open(new Path(path + fileName));
            IOUtils.copyBytes(inputStream, outputStream, 4 * 1024, true);
            result = true;
        } catch (IOException e) {
            logger.error("【下载文件失败】 ", e);

        } finally {
            IOUtils.closeStream(inputStream);
        }
        return result;
    }

    /**
     * 文件完整路径
     * @param path 文件前缀
     * @param fileName 文件名
     * @return 文件完整路径
     */
    public String getFilePath(String path, String fileName) {
        Path filePath = new Path(path + fileName);

        return hadoopProperties.getNameNode()+filePath.toString();
    }

    /**
     * 下载文件到本地
     *
     * @param path         文件路径
     * @param downloadPath 本地下载路径
     */
    public void downloadFileToLocal(String path, String downloadPath) {
        logger.info("【下载文件到本地】 开始下载, 文件路径: {}, 本地下载路径: {}", path, downloadPath);

        // 上传路径
        Path clientPath = new Path(path);
        // 目标路径
        Path serverPath = new Path(downloadPath);

        try {
            // 调用文件系统的文件复制方法，第一个参数是否删除原文件true为删除，默认为false
            fileSystem.copyToLocalFile(false, clientPath, serverPath);
        } catch (IOException e) {
            logger.error("【下载文件到本地失败】", e);
        }
    }

    /**
     * 获取目录信息
     *
     * @param path 目录路径
     * @return 目录信息
     */
    public List<Map<String, Object>> getPathInfo(String path) throws IOException {
        logger.info("【获取目录信息】 开始获取, 目录路径: {}", path);
        FileStatus[] statusList;
        List<Map<String, Object>> list = new ArrayList<>();

        statusList = fileSystem.listStatus(new Path(path));
        if (null != statusList && statusList.length > 0) {
            for (FileStatus fileStatus : statusList) {
                Map<String, Object> map = new HashMap<>(16);
                map.put("filePath", fileStatus.getPath());
                map.put("fileStatus", fileStatus.toString());
                list.add(map);
            }
        }
        return list;
    }

    /**
     * 获取目录下文件列表
     *
     * @param path 目录路径
     * @return List
     */
    public List<Map<String, String>> getFileList(String path) {
        logger.info("【获取目录下文件列表】 开始获取, 目录路径: {}", path);

        List<Map<String, String>> list = new ArrayList<>();
        try {
            // 递归找到所有文件
            RemoteIterator<LocatedFileStatus> filesList = fileSystem.listFiles(new Path(path), true);
            while (filesList.hasNext()) {
                LocatedFileStatus next = filesList.next();
                String fileName = next.getPath().getName();
                Path filePath = next.getPath();
                Map<String, String> map = new HashMap<>(16);
                map.put("fileName", fileName);
                map.put("filePath", filePath.toString());
                map.put("size", String.valueOf(next.getBlockSize()));
                map.put("owner", next.getOwner());
                list.add(map);
            }
        } catch (Exception e) {
            logger.error("【获取目录下文件列表失败】", e);
        }
        return list;
    }

    /**
     * 读取文件内容
     *
     * @param filePath 文件路径
     * @return
     */
    public String readFile(String filePath) {
        logger.info("【读取文件内容】 开始读取, 文件路径: {}", filePath);
        Path newPath = new Path(filePath);
        InputStream in = null;
        try {
            BufferedReader reader;
            StringBuilder buffer;
            buffer = new StringBuilder();
            in = fileSystem.open(newPath);
            // 用来保存每行读取的内容
            String line;
            // 设置字符编码，防止中文乱码
            reader = new BufferedReader(new InputStreamReader(in, "GBK"));
            line = reader.readLine();
            while (line != null) {
                buffer.append(line);
                buffer.append("\n");
                line = reader.readLine();
            }
            return buffer.toString();
        } catch (Exception e) {
            logger.error("【读取文件内容失败】", e);
        } finally {
            IOUtils.closeStream(in);
        }
        return null;
    }

    /**
     * 文件或文件夹重命名
     *
     * @param oldName 旧文件或旧文件夹名称
     * @param newName 新文件或新文件夹名称
     * @return 是否更改成功 true: 成功/false: 失败
     */
    public boolean renameFile(String oldName, String newName) {
        logger.info("【文件或文件夹重命名】 开始重命名, 旧文件或旧文件夹名称: {}, 新文件或新文件夹名称: {} ", oldName, newName);

        boolean isOk = false;
        Path oldPath = new Path(oldName);
        Path newPath = new Path(newName);
        try {
            isOk = fileSystem.rename(oldPath, newPath);
        } catch (IOException e) {

            logger.error("【文件或文件夹重命名失败】", e);
        }
        return isOk;
    }

    /**
     * 复制文件
     *
     * @param sourcePath 复制路径
     * @param targetPath 目标路径
     */
    public boolean copyFile(String sourcePath, String targetPath) {
        logger.info("【复制文件】 开始复制, 复制路径: {}, 目标路径: {}", sourcePath, targetPath);
        boolean result = false;
        // 原始文件路径
        Path oldPath = new Path(sourcePath);
        // 目标路径
        Path newPath = new Path(targetPath);
        FSDataInputStream inputStream;
        FSDataOutputStream outputStream;

        try {
            inputStream = fileSystem.open(oldPath);
            outputStream = fileSystem.create(newPath);
            IOUtils.copyBytes(inputStream, outputStream, 1024 * 1024 * 64, false);

            IOUtils.closeStream(inputStream);
            IOUtils.closeStream(outputStream);
            result = true;
        } catch (Exception e) {
            logger.error("【复制文件失败】", e);
        }
        return result;
    }
}
